package com.desheng.bigdata.flink.stream.transformation

import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

/**
  * connect
  */
object _04TransformationForConnectOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        val id2Score = env.fromCollection(List(
            Score(2, "语文", 80),
            Score(3, "数学", 70.5),
            Score(1, "英语", 30.5),
            Score(4, "体育", 60),
            Score(105, "信息技术", 100)
        ).map(score => (score.sid, score)))

        val id2Stu = env.fromCollection(List(
            Student(1, "霍龙飞", 18, 0),
            Student(2, "傅鑫达", 19, 1),
            Student(3, "刘卓凡", 20, 0),
            Student(4, "刘博", 20, 1),
            Student(106, "刘武", 30, 1)
        ).map(stu => (stu.id, stu)))

        val connected: ConnectedStreams[(Int, Score), (Int, Student)] = id2Score.connect(id2Stu)

//        connected.map(new CoMapFunction[(Int, Score), (Int, Student), String] {
//            override def map2(value: (Int, Student)): String = {
//                "student: >>>" + value._1 + "-->" + value._2.toString
//            }
//
//            override def map1(value: (Int, Score)): String = {
//                "score: >>>" + value._1 + "-->" + value._2.toString
//            }
//        }).print()

        connected.map(id2s => id2s._2.toString, id2sc => id2sc._2.toString).print()


        env.execute(s"${_04TransformationForConnectOps.getClass.getSimpleName}")
    }
}
case class Student(id: Int, name: String, age: Int, gender: Int)
case class Score(sid: Int, course: String, score: Double)
